package com.zc.redisstream.config;

import com.zc.config.FastJson2JsonRedisHashValueSerializer;
import com.zc.domain.ElectricAlarm;
import com.zc.domain.EnergyAlarm;
import com.zc.redisstream.listener.ElectricAlarmStreamListener;
import com.zc.redisstream.listener.EnergyAlarmStreamListener;
import com.zc.redisstream.listener.PhoenixAlarmStreamListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnWebApplication;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.connection.stream.*;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.StreamOperations;
import org.springframework.data.redis.serializer.RedisSerializer;
import org.springframework.data.redis.stream.StreamMessageListenerContainer;

import java.time.Duration;
import java.util.Collections;
import java.util.concurrent.Executor;

/**
 * redis stream 配置（redis5.0以上）
 */
@Configuration
@ConditionalOnWebApplication()
@SuppressWarnings({"rawtypes", "unchecked"})
public class RedisStreamConfig {

    private final Logger logger = LoggerFactory.getLogger(RedisStreamConfig.class);

    //指标阈值
    @Value("${redisStream.electricAlarmKey}")
    private String electricAlarmKey;

    //用能阈值
    @Value("${redisStream.energyAlarmKey}")
    private String energyAlarmKey;

    @Value("${redisStream.phoenixAlarmKey}")
    private String phoenixAlarmKey;

    @Autowired
    private RedisTemplate redisTemplate;

    @Autowired
    private ElectricAlarmStreamListener electricAlarmStreamListener;

    @Autowired
    private EnergyAlarmStreamListener energyAlarmStreamListener;

    @Autowired
    private PhoenixAlarmStreamListener phoenixAlarmStreamListener;


    private final Executor emsThreadPoolTaskExecutor;

    public RedisStreamConfig(@Qualifier("emsThreadPoolTaskExecutor") Executor emsThreadPoolTaskExecutor) {
        this.emsThreadPoolTaskExecutor = emsThreadPoolTaskExecutor;
    }

    /**
     *
     * 消费组${electricAlarmGroup}，收到消息后自动确认，消息处理完之后进行删除
     * 后续添加新的队列，队列名称、分组名称、消费者名称参考本实现方式进行添加
     *
     * @return
     */
    @Bean(initMethod = "start", destroyMethod = "stop")
    public StreamMessageListenerContainer electricAlarmListenerContainer(RedisConnectionFactory factory) {
        StreamMessageListenerContainer.StreamMessageListenerContainerOptions<String, ObjectRecord<String, ElectricAlarm>> options =
                StreamMessageListenerContainer.StreamMessageListenerContainerOptions
                        .builder()
                        // Stream 中没有消息时，阻塞多长时间，需要比 `spring.redis.timeout` 的时间小, 设置为0会导致CPU飙升
                        .pollTimeout(Duration.ofSeconds(2))
                        // 一次最多获取多少条消息
                        .batchSize(10)
                        // 运行 Stream 的 poll task
                        .executor(emsThreadPoolTaskExecutor)
                        // 获取消息的过程或获取到消息给具体的消息者处理的过程中，发生了异常的处理
                        .errorHandler(e -> {
                            logger.error("electricAlarm RedisStream异常", e);
                        })
                        .keySerializer(RedisSerializer.string())
                        .hashKeySerializer(RedisSerializer.string())
                        // 设置hashValue的序列号器
                        .hashValueSerializer(RedisSerializer.json())
//                        .objectMapper(new Jackson2HashMapper(false))
                        .targetType(ElectricAlarm.class)
                        .build();
        StreamMessageListenerContainer<String, ObjectRecord<String, ElectricAlarm>> listenerContainer = StreamMessageListenerContainer.create(factory, options);

        String groupName = electricAlarmKey + "_group";
        StreamOperations streamOperations = redisTemplate.opsForStream();
        RecordId recordId = null;
        // 如果队列不存在，则创建队列
        if (Boolean.FALSE.equals(redisTemplate.hasKey(electricAlarmKey))) {
            recordId = streamOperations.add(electricAlarmKey, Collections.singletonMap("_up", "up"));
            // 删除创建队列时的测试消息
            streamOperations.delete(electricAlarmKey, recordId);
        }
        // 如果分组不存在，则创建分组
        StreamInfo.XInfoGroups groups = streamOperations.groups(electricAlarmKey);
        long groupCount = groups.stream().filter(xInfoGroup -> xInfoGroup.groupName().equals(groupName)).count();
        if (groupCount <= 0) {
            streamOperations.createGroup(electricAlarmKey, groupName);
        }

        Consumer consumer = Consumer.from(groupName, "consumer_" + System.currentTimeMillis());
        listenerContainer.receiveAutoAck(consumer, StreamOffset.create(electricAlarmKey, ReadOffset.lastConsumed()), electricAlarmStreamListener);
        return listenerContainer;
    }



    //    /**
//     * 订阅者3，消费组${energyAlarmGroup}，收到消息后自动确认，消息处理完之后进行删除
//     * 后续添加新的队列，队列名称、分组名称、消费者名称参考本实现方式进行添加
//     * @param streamMessageListenerContainer
//     * @return
//     */
    @Bean(initMethod = "start", destroyMethod = "stop")
    public StreamMessageListenerContainer<String, ?> energyAlarmListenerContainer(RedisConnectionFactory factory) {
        StreamMessageListenerContainer.StreamMessageListenerContainerOptions<String, ObjectRecord<String, EnergyAlarm>> options =
                StreamMessageListenerContainer.StreamMessageListenerContainerOptions
                        .builder()
                        // Stream 中没有消息时，阻塞多长时间，需要比 `spring.redis.timeout` 的时间小, 设置为0会导致CPU飙升
                        .pollTimeout(Duration.ofSeconds(2))
                        // 一次最多获取多少条消息
                        .batchSize(10)
                        // 运行 Stream 的 poll task
                        .executor(emsThreadPoolTaskExecutor)
                        // 获取消息的过程或获取到消息给具体的消息者处理的过程中，发生了异常的处理
                        .errorHandler(e -> {
                            logger.error("energyAlarm RedisStream异常", e);
                        })
                        .keySerializer(RedisSerializer.string())
                        .hashKeySerializer(RedisSerializer.string())
                        .hashValueSerializer(new FastJson2JsonRedisHashValueSerializer<>(Object.class))
//                        .objectMapper(new ObjectHashMapper())
                        .targetType(EnergyAlarm.class)
                        .build();
        StreamMessageListenerContainer<String, ObjectRecord<String, EnergyAlarm>> listenerContainer = StreamMessageListenerContainer.create(factory, options);


        String groupName = energyAlarmKey + "_group";
        StreamOperations streamOperations = redisTemplate.opsForStream();
        RecordId recordId = null;
        // 如果队列不存在，则创建队列
        if (Boolean.FALSE.equals(redisTemplate.hasKey(energyAlarmKey))) {
            recordId = streamOperations.add(energyAlarmKey, Collections.singletonMap("_up", "up"));
            // 删除创建队列时的测试消息
            streamOperations.delete(energyAlarmKey, recordId);

        }
        // 如果分组不存在，则创建分组
        StreamInfo.XInfoGroups groups = streamOperations.groups(energyAlarmKey);
        long groupCount = groups.stream().filter(xInfoGroup -> xInfoGroup.groupName().equals(groupName)).count();
        if (groupCount <= 0) {
            streamOperations.createGroup(energyAlarmKey, groupName);
        }
        Consumer consumer = Consumer.from(groupName, "consumer_" + System.currentTimeMillis());
        listenerContainer.receiveAutoAck(consumer, StreamOffset.create(energyAlarmKey, ReadOffset.lastConsumed()), energyAlarmStreamListener);
        return listenerContainer;
    }


    /**

     */
    /**
     * 可以同时支持 独立消费 和 消费者组 消费
     * <p>
     * 可以支持动态的 增加和删除 消费者
     * <p>
     * 消费组需要预先创建出来
     *
     * 订阅者1，消费组${phoenixAlarmGroup}，收到消息后自动确认，消息处理完之后进行删除
     * 后续添加新的队列，队列名称、分组名称、消费者名称参考本实现方式进行添加
     *
     *
     * @return StreamMessageListenerContainer
     */
    @Bean(initMethod = "start", destroyMethod = "stop")
    public StreamMessageListenerContainer<String, MapRecord<String, String, String>> phoenixAlarmListenerContainer(RedisConnectionFactory factory) {
        StreamMessageListenerContainer.StreamMessageListenerContainerOptions<String, MapRecord<String, String, String>> options =
                StreamMessageListenerContainer.StreamMessageListenerContainerOptions
                        .builder()
                        // Stream 中没有消息时，阻塞多长时间，需要比 `spring.redis.timeout` 的时间小, 设置为0会导致CPU飙升
                        .pollTimeout(Duration.ofSeconds(2))
                        // 一次最多获取多少条消息
                        .batchSize(10)
                        // 运行 Stream 的 poll task
                        .executor(emsThreadPoolTaskExecutor)
                        // 获取消息的过程或获取到消息给具体的消息者处理的过程中，发生了异常的处理
                        .errorHandler(e -> {
                            logger.error("phoenixAlarm RedisStream异常", e);
                        })
                        .build();
        StreamMessageListenerContainer<String, MapRecord<String, String, String>> listenerContainer = StreamMessageListenerContainer.create(factory, options);

        String groupName = phoenixAlarmKey + "_group";

        StreamOperations streamOperations = redisTemplate.opsForStream();
        RecordId recordId = null;
        // 如果队列不存在，则创建队列
        if (Boolean.FALSE.equals(redisTemplate.hasKey(phoenixAlarmKey))) {
            recordId = streamOperations.add(phoenixAlarmKey, Collections.singletonMap("_up", "up"));
            // 删除创建队列时的测试消息
            streamOperations.delete(phoenixAlarmKey, recordId);
        }
        // 如果分组不存在，则创建分组
        StreamInfo.XInfoGroups groups = streamOperations.groups(phoenixAlarmKey);
        long groupCount = groups.stream().filter(xInfoGroup -> xInfoGroup.groupName().equals(groupName)).count();
        if (groupCount <= 0) {
            streamOperations.createGroup(phoenixAlarmKey, groupName);
        }

        Consumer consumer = Consumer.from(groupName, "consumer_" + System.currentTimeMillis());
        listenerContainer.receiveAutoAck(consumer, StreamOffset.create(phoenixAlarmKey, ReadOffset.lastConsumed()), phoenixAlarmStreamListener);
        return listenerContainer;
    }

}
